package com.atguigu.gulimall.search.thread;

import java.util.concurrent.*;

/**
 * @program: gulimall
 * @description:
 * @author: 江天赐
 * @create: 2022-08-09 15:32
 */
public class ThreadTest {
    public static ExecutorService service = Executors.newFixedThreadPool(10);

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        System.out.println("main...start...");
        /**
         * 方法完成后的感知
         */
//        CompletableFuture<Void> voidCompletableFuture = CompletableFuture.runAsync(() -> {
//            System.out.println("Thread01当前线程：" + Thread.currentThread().getId());
//            int i = 10 / 2;
//            System.out.println("运行结果：" + i);
//        }, service);
//        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
//            System.out.println("Thread01当前线程：" + Thread.currentThread().getId());
//            int i = 10 / 0;
//            System.out.println("运行结果：" + i);
//            return i;
//        }, service).whenComplete((res, exception) -> {
//            //能得到异常信息，但是没法修改返回数据
//            System.out.println("异步任务成功。。。结果是：" + res + "；异常是：" + exception);
//        }).exceptionally(throwable -> {
//            //出现异常给返回值设置默认值
//            return 10;
//        });
//        Integer integer = future.get();
//        System.out.println("使用future.get()获取的结果:" + integer);
        /**
         * 方法执行完成后的处理
         */
//        CompletableFuture<Integer> handle = CompletableFuture.supplyAsync(() -> {
//            System.out.println("Thread01当前线程：" + Thread.currentThread().getId());
//            int i = 10 / 4;
//            System.out.println("运行结果：" + i);
//            return i;
//        }, service).handle((res, thr) -> {
//            if (res != null) {
//                return res * 2;
//            }
//            if (thr != null) {
//                return 0;
//            }
//            return res * 3;
//        });
//        System.out.println("使用handle()获取结果："+handle.get());

        /**
         *线程串行化
         * 1)、thenRun:不能获取上一步的结果
         * .thenRunAsync(() -> {
         *             System.out.println("任务2启动了");
         *         }, service)
         *  2)、thenAccept能接受上一步的结果，无返回值
         *  .thenAcceptAsync((res) -> {
         *             System.out.println("任务2启动了");
         *         }, service);
         *   3)、thenApplyAsync能接受上一步的结果，有返回值
         *   .thenApplyAsync((res) -> {
         *             System.out.println("任务2启动了" + res);
         *             return "Hello" + res;
         *         }, service);
         */
//        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
//            System.out.println("Thread01当前线程：" + Thread.currentThread().getId());
//            int i = 10 / 4;
//            System.out.println("运行结果：" + i);
//            return i;
//        }, service).thenApplyAsync((res) -> {
//            System.out.println("任务2启动了" + res);
//            return "Hello" + res;
//        }, service);
//        System.out.println(future.get());

        /**
         * 两个任务必须都完成，触发该任务。:
         * thenCombine：组合两个 future，获取两个 future 的返回结果，并返回当前任务的返回值
         * thenAcceptBoth：组合两个 future，获取两个 future 任务的返回结果，然后处理任务，没有返回值。
         * runAfterBoth：组合两个 future，不需要获取 future 的结果，只需两个 future 处理完任务后， 处理该任务。
         */
        CompletableFuture<Object> future1 = CompletableFuture.supplyAsync(() -> {
            System.out.println("任务1线程：" + Thread.currentThread().getId());
            int i = 10 / 4;
            System.out.println("任务1结束：" + i);
            return i;
        }, service);

        CompletableFuture<Object> future2 = CompletableFuture.supplyAsync(() -> {
            System.out.println("任务2线程：" + Thread.currentThread().getId());
            int i = 10 / 4;
            try {
                Thread.sleep(5000);
                System.out.println("任务2结束：" + i);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "Hello";
        }, service);
//        future1.runAfterBothAsync(future2,()->{
//            System.out.println("任务3开始。。。");
//        },service);
//        future1.thenAcceptBothAsync(future2, (res1, res2) -> {
//            System.out.println("任务3开始;res1:" + res1 + ";res2:" + res2);
//        }, service);
//        CompletableFuture<String> stringCompletableFuture = future1.thenCombineAsync(future2, (res1, res2) -> {
//            System.out.println("任务4开始;res1:" + res1 + ";res2:" + res2);
//            return res1 + ":" + res2;
//        }, service);
//        System.out.println("任务4返回的结果：" + stringCompletableFuture);
        /**
         * 当两个任务中，任意一个 future 任务完成的时候，执行任务。
         * applyToEither：两个任务有一个执行完成，获取它的返回值，处理任务并有新的返回值。
         * acceptEither：两个任务有一个执行完成，获取它的返回值，处理任务，没有新的返回值。
         * runAfterEither：两个任务有一个执行完成，不需要获取 future 的结果，处理任务，也没有返 回值。
         */
//        future1.runAfterEitherAsync(future2,()->{
//            System.out.println("任务5开始");
//        },service);

//        future1.acceptEitherAsync(future2, (res) -> {
//            System.out.println("任务6开始" + res);
//        }, service);

//        CompletableFuture<String> stringCompletableFuture = future1.applyToEitherAsync(future2, (res) -> {
//            System.out.println("任务7开始" + res);
//            return "任务7:" + res;
//        }, service);
//        System.out.println(stringCompletableFuture.get());

        /**
         * 多任务组合
         * allOf：等待所有任务完成
         * allOf.get();//等待所有结果完成 阻塞
         * anyOf：只要有一个任务完成
         * anyOf.get() 最先完成的future的数据
         */
        CompletableFuture<String> futureImg = CompletableFuture.supplyAsync(() -> {
            System.out.println("查询商品的图片信息");
            return "hello.jpg";
        });

        CompletableFuture<String> futureAttr = CompletableFuture.supplyAsync(() -> {
            System.out.println("查询商品的属性信息");
            return "黑色+256G";
        });
        CompletableFuture<String> futureDesc = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(3000);
                System.out.println("查询商品的介绍");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "华为";
        });
        CompletableFuture<Void> allOf = CompletableFuture.allOf(futureImg, futureAttr, futureDesc);
        allOf.get();//等待所有结果完成
//        CompletableFuture<Object> anyOf = CompletableFuture.anyOf(futureImg, futureAttr, futureDesc);
//        System.out.println(anyOf.get());//anyOf.get() 最先完成的future的数据
        System.out.println("main...end...");
    }


    /**
     * CompletableFuture提供四个静态方法来创建一个异步操作
     * runXxxx都是没有返回结果的，需要Runnable做参数，supplyXxxx都是可以获取返回结果的(使用get方法)，需要Supplier<U>做参数
     * 可以传入自定义的线程池，否则就用默认的线程池
     * public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) {
     * return asyncSupplyStage(asyncPool, supplier);
     * }
     * <p>
     * public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier,
     * Executor executor) {
     * return asyncSupplyStage(screenExecutor(executor), supplier);
     * }
     * <p>
     * public static CompletableFuture<Void> runAsync(Runnable runnable) {
     * return asyncRunStage(asyncPool, runnable);
     * }
     * <p>
     * public static CompletableFuture<Void> runAsync(Runnable runnable,
     * Executor executor) {
     * return asyncRunStage(screenExecutor(executor), runnable);
     * }
     * =========================计算完成时回调方法=================================
     * public CompletionStage<T> whenComplete
     * (BiConsumer<? super T, ? super Throwable> action);
     * <p>
     * public CompletionStage<T> whenCompleteAsync
     * (BiConsumer<? super T, ? super Throwable> action);
     * <p>
     * public CompletionStage<T> whenCompleteAsync
     * (BiConsumer<? super T, ? super Throwable> action,
     * Executor executor);
     * <p>
     * public CompletionStage<T> exceptionally
     * (Function<Throwable, ? extends T> fn);
     * <p>
     * whenComplete 可以处理正常和异常的计算结果，exceptionally 处理异常情况。
     * whenComplete和 whenCompleteAsync 的区别： whenComplete：是执行当前任务的线程执行继续执行 whenComplete 的任务。
     * whenCompleteAsync：是执行把 whenCompleteAsync 这个任务继续提交给线程池 来进行执行。 方法不以 Async 结尾，
     * 意味着 Action 使用相同的线程执行，而 Async 可能会使用其他线程 执行（如果是使用相同的线程池，也可能会被同一个线程选中执行）
     * exceptionally：出现异常给返回值设置默认值
     * ==============================handle方法==================================
     * <p>
     * handle和 complete 一样，可对结果做最后的处理（可处理异常），可改变返回值。
     * handleAsync异步执行即新开一个线程
     * public <U> CompletableFuture<U> handle(
     * BiFunction<? super T, Throwable, ? extends U> fn) {
     * return uniHandleStage(null, fn);
     * }
     * <p>
     * public <U> CompletableFuture<U> handleAsync(
     * BiFunction<? super T, Throwable, ? extends U> fn) {
     * return uniHandleStage(asyncPool, fn);
     * }
     * <p>
     * public <U> CompletableFuture<U> handleAsync(
     * BiFunction<? super T, Throwable, ? extends U> fn, Executor executor) {
     * return uniHandleStage(screenExecutor(executor), fn);
     * }
     * ================线程串行化方法===================
     * public <U> CompletableFuture<U> thenApply(
     * Function<? super T,? extends U> fn) {
     * return uniApplyStage(null, fn);
     * }
     * <p>
     * public <U> CompletableFuture<U> thenApplyAsync(
     * Function<? super T,? extends U> fn) {
     * return uniApplyStage(asyncPool, fn);
     * }
     * <p>
     * public <U> CompletableFuture<U> thenApplyAsync(
     * Function<? super T,? extends U> fn, Executor executor) {
     * return uniApplyStage(screenExecutor(executor), fn);
     * }
     * <p>
     * public CompletableFuture<Void> thenAccept(Consumer<? super T> action) {
     * return uniAcceptStage(null, action);
     * }
     * <p>
     * public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action) {
     * return uniAcceptStage(asyncPool, action);
     * }
     * <p>
     * public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action,
     * Executor executor) {
     * return uniAcceptStage(screenExecutor(executor), action);
     * }
     * <p>
     * public CompletableFuture<Void> thenRun(Runnable action) {
     * return uniRunStage(null, action);
     * }
     * <p>
     * public CompletableFuture<Void> thenRunAsync(Runnable action) {
     * return uniRunStage(asyncPool, action);
     * }
     * <p>
     * public CompletableFuture<Void> thenRunAsync(Runnable action,
     * Executor executor) {
     * return uniRunStage(screenExecutor(executor), action);
     * }
     * <p>
     * thenApply 方法：当一个线程依赖另一个线程时，获取上一个任务返回的结果，并返回当前任务的返回值。
     * thenAccept 方法：消费处理结果。接收任务的处理结果，并消费处理，无返回结果。获取上一个任务返回的结果，不返回当前任务的返回值。
     * thenRun 方法：只要上面的任务执行完成，就开始执行 thenRun，只是处理完任务后，执行 thenRun 的后续操作
     * 带有 Async 默认是异步执行的。同之前。 以上都要前置任务成功完成。
     * Function<? super T,? extends U> T：上一个任务返回结果的类型 U：当前任务的返回值类型
     */


    public static void thread(String[] args) {
        System.out.println("main...start...");
        /**
         * 1.继承Thread
         *         Thread01 thread01 = new Thread01();
         *         thread01.start();
         * 2.实现Runnable接口
         *         Runnable01 runnable01 = new Runnable01();
         *         new Thread(runnable01).start();
         * 3.实现Callable接口+FutureTask(可以拿到返回结果，处理异常) call方法
         *         Callable01 callable01 = new Callable01();
         *         FutureTask<Integer> integerFutureTask = new FutureTask<>(callable01);
         *         new Thread(integerFutureTask).start();
         *         Integer integer = integerFutureTask.get();//阻塞等待整个线程完成，获取结果
         *         //调用FutureTask的get方法获取Callable的call方法返回的值
         *         FutureTask<V> implements RunnableFuture<V>
         *         RunnableFuture<V> extends Runnable, Future<V>//FutureTask -> 间接实现了Runnable接口，有run()
         *         FutureTask(Callable<V> callable)
         *         FutureTask(Runnable runnable, V result)//可以使用result来接受Runnable的值
         *  4.线程池[ExecutorService]：业务代码推荐使用线程池，将所有的多线程异步任务交给线程池使用，资源控制
         *      给线程池直接提交任务
         *      1.线程池的创建
         *          1.1）Executors.newFixedThreadPool 不建议使用
         *          1.2）new ThreadPoolExecutor
         *
         *          Future:可以获取异步结果
         *
         *  区别：1、2不能获取返回值。3可以获取返回值
         *       1、2、3都不能实现资源管理，只有4能，性能稳定。
         */
        //当前系统中，线程池最好保持一两个，异步任务提交给线程池执行
        // submit()有返回值、能传Callable、Runnable;execute无返回值、只能传Runnable;
        //都是向线程池异步提交任务
        //<T> Future<T> submit(Callable<T> task);
        //<T> Future<T> submit(Runnable task, T result);
        //Future<?> submit(Runnable task);
        //void execute(Runnable command);
        service.execute(new Runnable01());
/**七大核心线程池参数
 *      int corePoolSize：5,
 * //核心线程数，线程池创建还就准备就绪的线程数量，等待来接受异步任务执行，只要
 * 只要线程池不销毁，一直存在(除非设置了allowCoreThreadTimeOut)
 *      int maximumPoolSize,//最大线程数量，控制资源并发
 *      long keepAliveTime,//存活时间，如果当前的最大线程数量大于core数量且阻塞队列满了
 *          释放空闲的线程(maximumPoolSize-corePoolSize)即非核心线程，只要线程空闲大于指定的keepAliveTime，
 *          如果有新任务提交，它会处理新提交的任务，而不是阻塞队列中的
 *      TimeUnit unit,//时间单位
 *      BlockingQueue<Runnable> workQueue,//阻塞队列，如果任务有很多，就会将当前多的任务放到阻塞队列中
 *      //只要有线程空闲了，就会取队列取任务继续执行
 *      ThreadFactory threadFactory,//线程创建工厂
 *      RejectedExecutionHandler handler//如果阻塞队列满了，按照我们指定的拒绝策略执行任务
 *      new ThreadPoolExecutor.AbortPolicy() 默认的拒绝策略 这种拒绝策略在拒绝任务时，
 *      会直接抛出异常 RejectedExecutionException （属于RuntimeException），让你感知到任务被拒绝了，于是你便可以根据业务逻辑选择重试或者放弃提交等策略。
 *      new ThreadPoolExecutor.DiscardPolicy() 当新任务被提交后直接被丢弃掉，也不会给你任何的通知，相对而言存在一定的风险，因为我们提交的时候根本不知道这个任务会被丢弃，可能造成数据丢失。
 *      new ThreadPoolExecutor.DiscardOldestPolicy() 如果线程池没被关闭且没有能力执行，则会丢弃任务队列中的头结点，通常是存活时间最长的任务，这种策略与第二种不同之处在于它丢弃的不是最新提交的，
 *      而是队列中存活时间最长的，这样就可以腾出空间给新提交的任务，但同理它也存在一定的数据丢失风险。
 *      new ThreadPoolExecutor.CallerRunsPolicy()相对而言它就比较完善了，当有新任务提交后，如果线程池没被关闭且没有能力执行，则把这个任务交于提交任务的线程执行，也就是谁提交任务，谁就负责执行任务。这样做主要有两点好处。
 *      第一点新提交的任务不会被丢弃，这样也就不会造成业务损失。
 *      第二点好处是，由于谁提交任务谁就要负责执行任务，这样提交任务的线程就得负责执行任务，而执行任务又是比较耗时的，
 *      在这段期间，提交任务的线程被占用，也就不会再提交新的任务，减缓了任务提交的速度，相当于是一个负反馈。在此期间，
 *      线程池中的线程也可以充分利用这段时间来执行掉一部分任务，腾出一定的空间，相当于是给了线程池一定的缓冲期。
 *
 *      一个线程池：core：7，max:20,queue:50,100并发怎么分配
 *      7个core会立即执行，50个会进入阻塞队列，再开20-7=13个线程执行，剩下的30个线程使用拒绝策略
 *      new ThreadPoolExecutor.CallerRunsPolicy() 不想抛弃还要执行的话，采用这种拒绝策略
 *      常见的线程池
 *固定线程数的线程池 Executors.newFixedThreadPool(nThreads, nThreads,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>(),threadFactory)
 * 最大线程数就是核心线程数且一直存活               corePoolSize，maximumPoolSize，keepAliveTime   LinkedBlockingQueue无界队列 会出现OutOfMemoryError
 *可缓存线程池Executors.newCachedThreadPool(0, Integer.MAX_VALUE,60L, TimeUnit.SECONDS,new SynchronousQueue<Runnable>());
 * 核心线程数为0，所有都可以回收                  最大线程数：Integer.MAX_VALUE 当任务较多时，会创建很多的线程超过操作系统的上限
 *定时或周期性执行任务Executors.newScheduledThreadPool(int corePoolSize, ThreadFactory threadFactory)
 *使用唯一的线程去执行任务Executors.SingleThreadExecutor(1, 1,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>()))
 *                                                                                  LinkedBlockingQueue无界队列 会出现OutOfMemoryError
 *Executors.newSingleThreadScheduledExecutor()只是将ScheduledThreadPool 的核心线程数设置为了 1。 DelayedWorkQueue 延迟队列，同时也是一个无界队列会出现OutOfMemoryError
 *     public static ScheduledExecutorService newSingleThreadScheduledExecutor() {
 *         return new DelegatedScheduledExecutorService
 *             (new ScheduledThreadPoolExecutor(1));
 *     }
 *    public ScheduledThreadPoolExecutor(int corePoolSize) {
 *         super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
 *               new DelayedWorkQueue());
 *     }
 * ForkJoinPool第一点它非常适合执行可以产生子任务的任务。非常适合用于递归的场景，例如树的遍历、最优路径搜索等场景。
 * 第一步是拆分也就是 Fork，第二步是汇总也就是 Join
 *第二点不同之处在于内部结构，之前的线程池所有的线程共用一个队列，
 * 但 ForkJoinPool 线程池中每个线程都有自己独立的任务队列双端队列 deque，
 * 这时一旦线程中的任务被 Fork 分裂了，分裂出来的子任务放入线程自己的 deque 里
 *可以直接在自己的任务队列中获取而不必去公共队列中争抢也不会发生阻塞（除了后面会讲到的 steal 情况外），减少了线程间的竞争和切换，是非常高效的。
 * work-stealing：而线程 t1 的任务特别繁重，分裂了数十个子任务，但是 t0 此时却无事可做，
 * 它自己的 deque 队列为空，这时为了提高效率，t0 就会想办法帮助 t1 执行任务；
 * 线程 t1 获取任务的逻辑是后进先出，也就是LIFO，
 * 而线程 t0 在“steal”偷线程 t1 的 deque 中的任务的逻辑是先进先出，也就是FIFO
 * 使用 “work-stealing” 算法和双端队列很好地平衡了各线程的负载。
 *
 *   工作顺序
 *      1）、线程池创建，准别好core的核心线程，准备接受任务
 *      1.1)、core线程满了，就将再进来的任务放入阻塞队列中，空闲的core就会自己去阻塞队列中获取任务
 *      1.2）、阻塞队列满了，就直接开新线程执行，最大只能开到max指定的数量
 *      1.3）、max满了，就使用RejectedExecutionHandler拒绝策略
 *      1.4）、max都执行完成了，有很多空闲，在指定的时间(keepAliveTime)后,就会释放maximumPoolSize-corePoolSize这些线程
 *
 *      new LinkedBlockingDeque()默认是Integer的最大值。导致内存不够,进行压力测试测出合适的数据
 */

        ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 200, 10, TimeUnit.SECONDS, new LinkedBlockingDeque(10000), Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy());
        Thread01 thread01 = new Thread01();
        thread01.start();
        Runnable01 runnable01 = new Runnable01();
        new Thread(runnable01).start();
        Callable01 callable01 = new Callable01();
        FutureTask<Integer> integerFutureTask = new FutureTask<>(callable01);
        new Thread(integerFutureTask).start();
        try {
            //阻塞等待整个线程执行完成，获取返回结果
            Integer integer = integerFutureTask.get();
            System.out.println("Callable01的call返回的结果：" + integer);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
        System.out.println("main...end...");
    }

    public static class Thread01 extends Thread {
        @Override
        public void run() {
            System.out.println("Thread01当前线程：" + Thread.currentThread().getId());
            int i = 10 / 2;
            System.out.println("运行结果：" + i);
        }
    }

    public static class Runnable01 implements Runnable {
        @Override
        public void run() {
            System.out.println("Runnable01当前线程：" + Thread.currentThread().getId());
            int i = 10 / 2;
            System.out.println("运行结果：" + i);
        }
    }

    public static class Callable01 implements Callable<Integer> {
        @Override
        public Integer call() throws Exception {
            System.out.println("Callable01当前线程：" + Thread.currentThread().getId());
            int i = 10 / 2;
            return i;
        }
    }

}
